package com.jfire.socket.socketclient;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.jfire.baseutil.collection.ByteBufferPool;
import com.jfire.baseutil.collection.ByteCache;
import com.jfire.baseutil.collection.ByteCachePool;
import com.jfire.socket.socketclient.filter.AesFilter;
import com.jfire.socket.socketclient.filter.DataFilter;
import com.jfire.socket.socketclient.listen.ClientChannelInfo;
import com.jfire.socket.socketclient.listen.ClientReadHandler;
import com.jfire.socket.socketclient.listen.GetReadResult;
import com.jfire.socket.socketclient.listen.ReadResult;
import com.jfire.socket.socketserver.bus.ActionType;
import com.jfire.socket.socketserver.bus.BusinessCommand;
import com.jfire.socket.socketserver.exception.ConnectErrorException;
import com.jfire.socket.socketserver.util.DefaultHeadFactory;
import com.jfire.socket.socketserver.util.HeadFactory;

/**
 * 客户端工具类，注意，该客户端是非线程安全类，其方法不可多线程运行
 * 
 * @author linbin
 * 
 */
public class Client
{
    protected long              readTimeout         = 3000;
    protected long              reuseChannelTimeout = 60000;
    protected ClientChannelInfo clientChannelInfo;
    protected String            ip;
    protected int               port;
    protected ByteBuffer        writeBuffer         = ByteBufferPool.getBuffer(1024);
    protected DataFilter[]      dataFilters         = new DataFilter[0];
    protected int               filterNum           = 0;
    protected AesFilter         aesFilter;
    protected GetReadResult     getReadResult;
    protected ReentrantLock     lock                = new ReentrantLock();
    protected Condition         finishRead          = lock.newCondition();
    protected final Logger      logger              = LogManager.getLogger();
    protected int               connectTrytimes     = 4;
    protected HeadFactory       headFactory         = new DefaultHeadFactory();
    
    public Client(GetReadResult getReadResult)
    {
        this.getReadResult = getReadResult;
    }
    
    public Client(GetReadResult getReadResult, HeadFactory headFactory)
    {
        this.getReadResult = getReadResult;
        this.headFactory = headFactory;
    }
    
    /**
     * 根据命令和内容缓存，组成完成的报文发送，并且返回一个异步调用的future对象
     * 
     * @param command
     * @param cache
     * @throws ConnectErrorException
     * @throws IOException
     */
    public Future<Object> sendData(byte command, ByteCache cache) throws ConnectErrorException
    {
        getClientChannelInfo(true);
        ByteCache buckup = ByteCachePool.get();
        buckup.putByteCache(cache);
        while (true)
        {
            try
            {
                // 数据准备必须放在循环里面，并且放在发送数据的开头。
                // 因为如果一旦通道异常需要关闭，那么发送数据的过滤过程就需要重新来一次（比如加密的密钥不同了，就需要重新加密）
                // 并且此时的数据必须是原始数据
                AsynchronousSocketChannel socketChannel = clientChannelInfo.getSocketChannel();
                prepareData(command, cache);
                if (logger.isTraceEnabled())
                {
                    logger.trace("客户端{}数据准备发送长度为{}的数据", clientChannelInfo.getAddress(), writeBuffer.remaining());
                }
                ReadResult future = new ReadResult(lock, finishRead, clientChannelInfo);
                clientChannelInfo.offerReadResult(future);
                while (writeBuffer.hasRemaining())
                {
                    socketChannel.write(writeBuffer).get();
                }
                ByteCachePool.returnCache(buckup);
                return future;
            }
            catch (Exception e)
            {
                getClientChannelInfo(false);
                // 一旦重新获取通道，由于之前准备数据的时候已经对数据进行了过滤处理，所以此时需要将cache清空并且重新放上最初的原始数据
                // 否则再次准备数据的时候就会造成重复过滤或加密的问题
                cache.clear().putByteCache(buckup);
                logger.debug("获取新通道，重新发送");
                continue;
            }
        }
    }
    
    /**
     * 将需要发送的数据以及指令写入发送的的buffer
     * 
     * @param command 发送的内容的类型
     * @param cache 发送的具体内容
     */
    protected void prepareData(byte command, ByteCache cache)
    {
        if (cache == null)
        {
            writeBuffer.clear();
            headFactory.putHeadInBuffer(writeBuffer, command, 0);
            writeBuffer.put(ActionType.CLIENT_SEND).put(command).put(BusinessCommand.REQUEST_SUCCESS);
            writeBuffer.putInt(0);
            writeBuffer.flip();
            return;
        }
        for (int i = filterNum - 1; i > -1; i--)
        {
            dataFilters[i].outFilter(cache);
        }
        if (aesFilter != null)
        {
            aesFilter.outFilter(cache);
        }
        if (cache.remaining() > writeBuffer.capacity() - 8)
        {
            writeBuffer = ByteBufferPool.expandToSize(writeBuffer.capacity() + cache.remaining(), writeBuffer);
        }
        writeBuffer.clear();
        headFactory.putHeadInBuffer(writeBuffer, command, cache.getCount());
        writeBuffer.put(ActionType.CLIENT_SEND).put(command).put(BusinessCommand.REQUEST_SUCCESS);
        writeBuffer.putInt(cache.getCount());
        writeBuffer.put(cache.getDirectArray(), 0, cache.getCount());
        writeBuffer.flip();
    }
    
    /**
     * 打开ip和端口对应的socket通道。如果该通道已经被打开，并且还在有效期内，则直接使用该通道
     * 如果参数为true表示可以重用通道。否则需要重新获取
     * 
     * @param reuse 表示允许重用通道
     * @return
     * @throws ConnectErrorException
     */
    protected void getClientChannelInfo(boolean reuse) throws ConnectErrorException
    {
        if (reuse && clientChannelInfo != null && clientChannelInfo.expired() == false)
        {
            return;
        }
        if (clientChannelInfo != null)
        {
            clientChannelInfo.closeSocket();
        }
        boolean finishConnect = false;
        int tryTime = connectTrytimes;
        String address = null;
        AsynchronousSocketChannel socketChannel = null;
        do
        {
            try
            {
                socketChannel = AsynchronousSocketChannel.open(ChannelGroupSource.getChannelGroup());
                socketChannel.connect(new InetSocketAddress(ip, port)).get(5, TimeUnit.SECONDS);
                finishConnect = true;
                address = socketChannel.getLocalAddress().toString() + "--->" + socketChannel.getRemoteAddress().toString();
                break;
            }
            catch (Exception e)
            {
                try
                {
                    if (socketChannel != null)
                    {
                        socketChannel.close();
                    }
                }
                catch (Exception e1)
                {
                    ;
                }
                tryTime--;
            }
        } while (tryTime > 0);
        if (finishConnect == false)
        {
            throw new ConnectErrorException("经过" + connectTrytimes + "次测试，无法完成连接");
        }
        clientChannelInfo = buildChannelInfo(address, socketChannel);
        clientChannelInfo.startReadWait();
    }
    
    protected ClientChannelInfo buildChannelInfo(String address, AsynchronousSocketChannel socketChannel)
    {
        ClientReadHandler clientReadHandler = new ClientReadHandler(headFactory);
        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(clientReadHandler);
        clientChannelInfo.setClientUtil(this);
        clientChannelInfo.setLock(lock);
        clientChannelInfo.setFinishRead(finishRead);
        clientChannelInfo.setGetReadResult(getReadResult);
        clientChannelInfo.setAddress(address);
        clientChannelInfo.setReadTimeout(readTimeout);
        clientChannelInfo.setReuseChannelTimeout(reuseChannelTimeout);
        clientChannelInfo.setSocketChannel(socketChannel);
        clientChannelInfo.setDataFilters(dataFilters);
        clientChannelInfo.setAesFilter(aesFilter);
        clientChannelInfo.setLastUseChannelTime();
        return clientChannelInfo;
    }
    
    /**
     * 关闭客户端，并且关闭连接。归还缓存资源
     */
    public void close()
    {
        if (clientChannelInfo != null)
        {
            clientChannelInfo.closeSocket();
        }
        ByteBufferPool.returnBuffer(writeBuffer);
        writeBuffer = null;
    }
    
    /**
     * 关闭当前使用的通道
     * 如果客户端下次再发起数据发送请求，则会再次启动一个通道
     */
    public void closeCurrentChannel()
    {
        if (clientChannelInfo != null)
        {
            clientChannelInfo.closeSocket();
        }
    }
    
    public Client setReadTimeout(long readTimeout)
    {
        this.readTimeout = readTimeout;
        return this;
    }
    
    public Client setReuseChannelTimeout(long reuseChannelTimeout)
    {
        this.reuseChannelTimeout = reuseChannelTimeout;
        return this;
    }
    
    public Client setIp(String ip)
    {
        this.ip = ip;
        return this;
    }
    
    public Client setPort(int port)
    {
        this.port = port;
        return this;
    }
    
    /**
     * 设置数据过滤器。数据收到时，是从数组的开始过滤。数据发出时，是从数组的末尾往前进行过滤
     * 
     * @param dataFilters
     * @return
     */
    public Client setDataFilters(DataFilter... dataFilters)
    {
        if (dataFilters == null || dataFilters.length == 0)
        {
            return this;
        }
        this.dataFilters = dataFilters;
        filterNum = dataFilters.length;
        return this;
    }
    
    public Client setConnectTrytimes(int connectTrytimes)
    {
        this.connectTrytimes = connectTrytimes;
        return this;
    }
    
    public HeadFactory getHeadFactory()
    {
        return headFactory;
    }
    
    public Client setHeadFactory(HeadFactory headFactory)
    {
        this.headFactory = headFactory;
        return this;
    }
    
}
